# -*- coding: utf-8 -*-
# standard
import os
import sys
import time
import ctypes
import asyncio
import threading

# third
import httpx

# local
_P_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if _P_PATH not in sys.path:
    sys.path.append(_P_PATH)
from functions.tianqing_func import *


"""
@Title:  从第三方平台采集数据的脚本
@File: collect_func.py
@Author: walle 2024年11日08日 11时57分26秒
@Version: 1.0.0
@Desc: 
三相电系统中, 这三个相通常被称为A相、B相和C相, 
它们三个频率相同、振幅相等、相位依次互差120°的交流电势组成的电源供电。
在三相电系统中，为了区分和识别不同的相，通常会采用特定的颜色来标记。具体来说：
A相: 黄色
B相: 绿色
C相: 红色
相与相之间的电压被称为线电压,其电压值通常为380V。
而相与中心线(零线)之间的电压被称为相电压,其电压值通常为220V。

"""


async def tianqing_get_sensor_data() -> Resp:
    """
    获取天勍的温度/电量传感器数据, 这里是用于生产的异步函数，
    这里使用异步而不是多线程的目的是尽早的让出等待时间以避免对fastapi的
    事件循环占用过多的时间片。
    :return: _description_
    """
    sensors = tianqing_get_sensors_from_local()
    async with httpx.AsyncClient(timeout=15) as client:
        resp = tianqing_get_headers()
        if resp:
            headers = resp.data
        else:
            logger.error(f"获取天勍 headers 失败, error: {resp.message}")
            return  # 中止运行
        url = 'https://xcx.itianqing.com/scat-api/v1/wms/data/getDeviceData'
        for i, sensor in enumerate(sensors):
            sensor_category = sensor['sensor_category']
            if sensor_category == SensorCategory.ELECTRICITY.value:
                device_type = "4"
                measurement = SensorCategory.get_influxdb_measurement(SensorCategory.ELECTRICITY.value)
                name = "电量传感器"
                key = "et"
            else:
                device_type = "2"
                measurement = SensorCategory.get_influxdb_measurement(SensorCategory.TEMPERATURE.value)
                name = "温度传感器"
                key = 't'
            params = {
                "merchantId": sensor['merchant_no'],
                "hostNo": sensor['host_no'],
                "sensorId": sensor['sensor_sn'],
                "type": device_type,
            }
            try:
                response = await client.get(url=url, params=params, headers=headers)
            except Exception as e:
                logger.error(e)
                resp.message = f"请求天勍接口失败:{e}"
                response = None
            finally:
                if not resp:
                    pass
                elif response is None:
                    logger.error(f"resp.message = {resp.message}")
                elif response.status_code != 200:
                    logger.error(f"接口返回了错误的状态码: {response.status_code}")
                else:
                    result = response.json()
                    if result['status'] != 0:
                        logger.error(f"请求失败, error:{result['error']}, params: {params}")
                    else:
                        data = result['data']
                        if data is None:
                            logger.error(f"请求成功, 但是没有数据, params: {params}, result: {result}")
                            continue
                        """
                        {'lastTime': '2024-11-13 15:25:19', 
                        'hostNo': '98D8635379E4', 
                        'lastData': {
                                'eta': '1917', # A相电量
                                'etb': '1475', # B相电量                          
                                'etc': '1132', # C相电量
                                'et': '98066'  # 总电量
                                'va': '232',  # A相电压
                                'vb': '232',  # B相电压
                                'vc': '232', # C相电压
                                'ea': '20', # A相电流
                                'eb': '11', # B相电流
                                'ec': '11', # C相电流
                                'type': 'TBC', # TBC/多相电, SBC/单相电                      
                                }, 
                        'sensorId': '4A600AE4663E'}
                        """
                        customer_id = sensor['customer_id']
                        tags = {
                            "customer_id": customer_id,
                            "store_id": sensor['store_id'],
                            "uuid": sensor['host_no'],
                            "grp": 1,  # group_id
                            "id": sensor['sensor_sn'],
                        }
                        fields = {}
                        # 只取总功率(电量传感器)或者温度(温度传感器)
                        tags["type"] = data['lastData'].get("type")
                        et = data['lastData'].get("et")
                        if et:
                            fields = {key: float(et)}  
                        # 取全部的数据
                        # for k, v in data['lastData'].items():
                        #     if k == "type":  
                        #         tags[k] = v
                        #     elif v is None:
                        #         pass
                        #     else:
                        #         fields[k] = float(v) 
                        if fields:
                            bucket_name = generate_customer_bucket_name(customer_id)
                            result = tianqing_save_to_db(measurement=measurement,
                                                         bucket_name=bucket_name,
                                                         tags=tags, fields=fields)
                            if result == 0:
                                logger.debug(f"{name}数据写入influxdb成功")
                            else:
                                logger.error(f"{name}数据写入influxdb失败, measurement: {measurement}, bucket_name: {bucket_name}, tags:{tags}, fields: {fields}")
                        else:
                            logger.error(f"没有获取有效的传感器数据! data: {data}")






async def collect_third_party_platform_data() -> Resp:
    """
    采集第三方平台数据
    """
    resp = Resp()
    try:
        await tianqing_get_sensor_data()
    except Exception as e:
        logger.exception(e)
        resp.message = f"调用天勍后台采集函数失败:{e}"
    finally:
        return resp
    

######################## 节拍器 ##############################


class BeaterThread(threading.Thread):
    """
    节拍器线程
    """
    func: Callable | None = None
    interval: int  # 发送间隔，单位秒

    def __init__(self, minute: int = 3):
        super().__init__(name=self.get_thread_name(), daemon=True)
        self.func = self.send_collect_data_request
        logger.debug(f"节拍器间隔:{minute}分钟")
        self.interval = minute * 60

    def run(self):
        time.sleep(18)
        while True:
            self.send_collect_data_request()
            time.sleep(self.interval)

    @staticmethod
    def send_collect_data_request():
        url = "http://127.0.0.1:8085/collect/collect_now"
        try:
            response = requests.get(url=url)
            if response.status_code != 200:
                logger.error(f"本地服务器返回了错误的状态码:{response.status_code}")
        except Exception as e:
            logger.error(e)

    @classmethod
    def get_thread_name(cls) -> str:
        """
        获取线程名称
        :return: _description_
        """      
        return cls.__name__
    
    @classmethod
    def kill_old_thread(cls):
        """
        杀死旧线程， 
        """
        consume_thread_name = cls.get_thread_name()
        for t in threading.enumerate():
            if t.name == consume_thread_name:
                result = ctypes.pythonapi.PyThreadState_SetAsyncExc(t._ident, ctypes.py_object(SystemExit))  # 第一个参数时线程id，第二个参数时一个异常类的 py_object 对象
                # result 是整数， 是终结线程的数量。
                del t  # 手动清理对象更佳，之后t就变成未定义了。
                logger.warning(f"杀死消费线程 {consume_thread_name} 成功 {result}")
                break
    

#########################测试函数############################


"""
下面的测试结果是
* 异步发送10个请求总耗时:2.463628秒
* 多线程发送10个请求总耗时:0.506151秒
这并不能说明多线程更优秀，相反，得益于服务端的异步架构，得以快速的处理
多线程几乎同时发送的大量请求，你可以从多线程的处理时间互相重叠中得出结果。
而异步客户端的落后时因为发送请求时，异步客户端在依次发送请求导致。
"""

async def test_async_request_speed(n):
    """
    测试异步耗时的速度
    """
    url = "http://127.0.0.1:8085/collect/collect_now"
    async with httpx.AsyncClient() as client:
        b = datetime.now()
        for i in range(n):
            response = await client.get(url, params={"num": i})
            print(response.json())
        e = datetime.now()
        d = (e - b).total_seconds()
        print(f"异步发送{n}个请求总耗时:{d}秒")


def test_multi_thread_request_speed(n):
    from concurrent.futures import ThreadPoolExecutor
    from concurrent.futures import as_completed

    url = "http://127.0.0.1:8085/collect/collect_now"
    with ThreadPoolExecutor(max_workers=n) as pool:
        b = datetime.now()
        futures = []
        for i in range(n):
            future = pool.submit(requests.get, url=url, params={"num": i})
            futures.append(future)
        for future in as_completed(futures):
            print(future.result().json())
        e = datetime.now()
        d = (e - b).total_seconds()
        print(f"多线程发送{n}个请求总耗时:{d}秒")



if __name__ == '__main__':
    print("-------------异步---------------")
    asyncio.run(test_async_request_speed(10))
    print("-------------同步---------------")
    test_multi_thread_request_speed(10)
    # import time
    # for i in range(100):
    #     asyncio.run(tianqing_get_sensor_data())
    #     time.sleep(60 * 2)
    pass